package com.bs.elasticsearch.service.impl;

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.text.csv.CsvData;
import cn.hutool.core.text.csv.CsvReadConfig;
import cn.hutool.core.text.csv.CsvRow;
import cn.hutool.core.text.csv.CsvUtil;
import com.bs.common.utils.Assert;
import com.bs.common.utils.CommonUtil;
import com.bs.common.utils.PageUtil;
import com.bs.common.utils.SnowFlakeUtil;
import com.bs.elasticsearch.service.ReadCsvToElasticService;
import com.bs.elasticsearch.vo.*;
import com.bs.elasticsearch.vo.param.NearByPoiParam;
import com.bs.elasticsearch.vo.param.SearchNamePoiParam;
import lombok.extern.log4j.Log4j2;
import org.elasticsearch.common.geo.GeoDistance;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.geo.builders.PointBuilder;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.index.query.GeoDistanceQueryBuilder;
import org.elasticsearch.index.query.GeoShapeQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.sort.GeoDistanceSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.BeanUtils;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author bingshao
 * @date 2023/4/7
 **/
@Log4j2
@Service
public class ReadCsvToElasticServiceImpl implements ReadCsvToElasticService {

    @Resource
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Override
    public void createIndex() {
        IndexOperations indexOperations = elasticsearchRestTemplate.indexOps(LocationPoiVo.class);
        indexOperations.create();
        Document mapping = indexOperations.createMapping();
        indexOperations.putMapping(mapping);
    }

    @Override
    public void readCsv(MultipartFile file, AddressIdVo addressIdVo) throws IOException {
        List<LocationVo> locationVoList = new ArrayList<>();
        CsvReadConfig csvReadConfig = new CsvReadConfig();
        csvReadConfig.setContainsHeader(true);
        csvReadConfig.setSkipEmptyRows(false);
        CsvData read = CsvUtil.getReader(csvReadConfig).read(IoUtil.getReader(new InputStreamReader(file.getInputStream())));
        for (CsvRow csvRow : read) {
            LocationVo locationVo = new LocationVo();
            BeanUtils.copyProperties(addressIdVo, locationVo);
            locationVo.setId(SnowFlakeUtil.getId());
            locationVo.setName(csvRow.get(0));
            locationVo.setAddress(csvRow.get(3));
            locationVo.setPhone(csvRow.get(4));
            locationVo.setCatalog(csvRow.get(5));
            GeoPoint geoPoint = new GeoPoint(Double.parseDouble(csvRow.get(2)), Double.parseDouble(csvRow.get(1)));
            locationVo.setGeoPoint(geoPoint);
            locationVoList.add(locationVo);
        }
        elasticsearchRestTemplate.save(locationVoList);
    }

    @Override
    public PageUtil<List<LocationVo>> geoSearch(Long id) {
        List<LocationVo> locationVoList = new ArrayList<>();

        LocationVo locationVo = elasticsearchRestTemplate.get(id.toString(), LocationVo.class);

        // 分页，只展示前20条
        Pageable pageable = PageRequest.of(0, 20);

        // 搜索字段为 location
        GeoDistanceQueryBuilder geoBuilder = new GeoDistanceQueryBuilder("geoPoint");
        //指定从哪个位置搜索
        geoBuilder.point(locationVo.getGeoPoint().getLat(), locationVo.getGeoPoint().getLon());
        //指定搜索多少km
        geoBuilder.distance(5, DistanceUnit.KILOMETERS);

        // 距离排序
        GeoDistanceSortBuilder sortBuilder = new GeoDistanceSortBuilder("geoPoint", locationVo.getGeoPoint().getLat(), locationVo.getGeoPoint().getLon());
        //升序
        sortBuilder.order(SortOrder.ASC);

        SearchHits<LocationVo> searchHits = elasticsearchRestTemplate
                .search(new NativeSearchQueryBuilder()
                        .withPageable(pageable)
                        .withFilter(geoBuilder)
                        .withSort(sortBuilder)
                        .build(), LocationVo.class);
        searchHits.forEach(searchHit -> {
            double calculate = GeoDistance.PLANE.calculate(searchHit.getContent().getGeoPoint().getLat(), searchHit.getContent().getGeoPoint().getLon(), locationVo.getGeoPoint().lat(), locationVo.getGeoPoint().lon(), DistanceUnit.METERS);
            System.out.println(searchHit.getContent().getName() + "距离" + calculate + "米");
            locationVoList.add(searchHit.getContent());
        });
        return PageUtil.instance(searchHits.getTotalHits(), locationVoList);
    }

    @Override
    public GeoLocationAreaVo geoByLonAndLat(Double lon, Double lat) throws IOException {
        GeoLocationAreaVo geoLocationAreaVo = new GeoLocationAreaVo();
        List<String> areaNameList = new ArrayList<>();
        List<String> areaCodeList = new ArrayList<>();
        List<GeoLocationVo> geoLocationVoList = new ArrayList<>();
        GeoShapeQueryBuilder geoShapeQueryBuilder = QueryBuilders.geoShapeQuery("coordinates", PointBuilder.newPoint(lon, lat).buildGeometry());
        SearchHits<GeoLocationVo> searchHits = elasticsearchRestTemplate.search(new NativeSearchQueryBuilder()
                .withFilter(geoShapeQueryBuilder)
                .build(), GeoLocationVo.class);
        searchHits.getSearchHits().forEach(searchHit -> {
            GeoLocationVo geoLocationVo = searchHit.getContent();
            geoLocationVoList.add(geoLocationVo);
        });
        List<GeoLocationVo> geoLocationVos = geoLocationVoList.stream().sorted(Comparator.comparing(GeoLocationVo::getDeep)).collect(Collectors.toList());
        geoLocationVos.forEach(geoLocationVo -> {
            areaNameList.add(geoLocationVo.getName());
            areaCodeList.add(CommonUtil.formatNumber(geoLocationVo.getAreaId(), 9));
        });
        geoLocationAreaVo.setAreaNameList(areaNameList);
        geoLocationAreaVo.setAreaCodeList(areaCodeList);
        geoLocationAreaVo.setGeoWkt(geoLocationVos.get(geoLocationVos.size() - 1).getGeo_wkt());
        geoLocationAreaVo.setCoordinates(geoLocationVos.get(geoLocationVos.size() - 1).getCoordinates());
        return geoLocationAreaVo;
    }

    @Override
    public void readPoiCsv(MultipartFile file) throws Exception {
        List<LocationPoiVo> locationPoiVoList = new ArrayList<>();
        CsvReadConfig csvReadConfig = new CsvReadConfig();
        csvReadConfig.setContainsHeader(true);
        csvReadConfig.setSkipEmptyRows(false);
        CsvData read = CsvUtil.getReader(csvReadConfig).read(IoUtil.getReader(new InputStreamReader(file.getInputStream())));
        Iterator<CsvRow> iterator = read.iterator();
        while (iterator.hasNext()) {
            try {
                CsvRow csvRow = iterator.next();
                LocationPoiVo locationPoiVo = new LocationPoiVo();
                locationPoiVo.setId(SnowFlakeUtil.getId());
                locationPoiVo.setName(csvRow.get(0));
                locationPoiVo.setAddress(csvRow.get(3));
                locationPoiVo.setPhone(csvRow.get(4));
                locationPoiVo.setCatalog(csvRow.get(5));
                GeoPoint geoPoint = new GeoPoint(Double.parseDouble(csvRow.get(2)), Double.parseDouble(csvRow.get(1)));
                locationPoiVo.setGeoPoint(geoPoint);
                locationPoiVoList.add(locationPoiVo);
            } catch (Exception e) {
                log.info(e.getMessage());
                Assert.isError(e.getMessage());
            }

        }
        elasticsearchRestTemplate.save(locationPoiVoList);
    }

    @Override
    public List<GeoLocationPoiVo> getNearByPoi(NearByPoiParam nearByPoiParam) {
        List<GeoLocationPoiVo> geoLocationPoiVoList = new ArrayList<>();

        Pageable pageable = PageRequest.of(0, nearByPoiParam.getSize());

        // 搜索字段为 location
        GeoDistanceQueryBuilder geoBuilder = new GeoDistanceQueryBuilder("geoPoint");
        //指定从哪个位置搜索
        geoBuilder.point(nearByPoiParam.getLat(), nearByPoiParam.getLon());
        //指定搜索多少km
        geoBuilder.distance(nearByPoiParam.getDistance(), DistanceUnit.KILOMETERS);

        // 距离排序
        GeoDistanceSortBuilder sortBuilder = new GeoDistanceSortBuilder("geoPoint", nearByPoiParam.getLat(), nearByPoiParam.getLon());
        //升序
        sortBuilder.order(SortOrder.ASC);

        SearchHits<LocationPoiVo> searchHits = elasticsearchRestTemplate
                .search(new NativeSearchQueryBuilder()
                        .withPageable(pageable)
                        .withFilter(geoBuilder)
                        .withSort(sortBuilder)
                        .build(), LocationPoiVo.class);
        searchHits.forEach(searchHit -> {
            double calculate = GeoDistance.PLANE.calculate(searchHit.getContent().getGeoPoint().getLat(), searchHit.getContent().getGeoPoint().getLon(), nearByPoiParam.getLat(), nearByPoiParam.getLon(), DistanceUnit.METERS);
            GeoLocationPoiVo geoLocationPoiVo = new GeoLocationPoiVo();
            BeanUtils.copyProperties(searchHit.getContent(), geoLocationPoiVo);
//            try {
//                GeoLocationAreaVo geoLocationAreaVo = geoByLonAndLat(searchHit.getContent().getGeoPoint().getLon(), searchHit.getContent().getGeoPoint().getLat());
//                geoLocationPoiVo.setAreaCodeList(geoLocationAreaVo.getAreaCodeList());
//                geoLocationPoiVo.setAreaNameList(geoLocationAreaVo.getAreaNameList());
//                geoLocationPoiVo.setGeoWkt(geoLocationAreaVo.getGeoWkt());
//                geoLocationPoiVo.setCoordinates(geoLocationAreaVo.getCoordinates());
            geoLocationPoiVo.setSpecificDistance(calculate + "米");
//            } catch (IOException e) {
//                log.info(e.getMessage());
//                Assert.isError(e.getMessage());
//            }

            geoLocationPoiVoList.add(geoLocationPoiVo);
        });
        return geoLocationPoiVoList;
    }

    @Override
    public List<GeoLocationPoiVo> searchNamePoi(SearchNamePoiParam searchNamePoiParam) {
        List<GeoLocationPoiVo> geoLocationPoiVoList = new ArrayList<>();
        Pageable pageable = PageRequest.of(0, searchNamePoiParam.getSize());
        SearchHits<LocationPoiVo> searchHits = elasticsearchRestTemplate
                .search(new NativeSearchQueryBuilder().withQuery(QueryBuilders.matchQuery("name", searchNamePoiParam.getName()))
                        .withPageable(pageable)
                        .build(), LocationPoiVo.class);
        searchHits.forEach(searchHit -> {
            GeoLocationPoiVo geoLocationPoiVo = new GeoLocationPoiVo();
            BeanUtils.copyProperties(searchHit.getContent(), geoLocationPoiVo);
            geoLocationPoiVoList.add(geoLocationPoiVo);
        });
        return geoLocationPoiVoList;
    }

    @Override
    public List<GeoLocationVo> getByAreaCode(List<String> areaCodeList) {
        List<GeoLocationVo> geoLocationVoList = new ArrayList<>();
        areaCodeList = areaCodeList.stream().map(CommonUtil::areaCodeRemoveZeros).collect(Collectors.toList());
        SearchHits<GeoLocationVo> searchHits = elasticsearchRestTemplate.search(new NativeSearchQueryBuilder().withQuery(QueryBuilders.termsQuery("areaId", areaCodeList)).build(), GeoLocationVo.class);
        searchHits.forEach(searchHit -> {
            GeoLocationVo geoLocationVo = new GeoLocationVo();
            BeanUtils.copyProperties(searchHit.getContent(), geoLocationVo);
            geoLocationVoList.add(geoLocationVo);
        });
        return geoLocationVoList;
    }

}
